package com.xschen.zk.advanced.services;

import com.xschen.zk.advanced.common.Constants;
import org.apache.zookeeper.*;

import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * 3. 如何通过zookeeper实现服务注册中心
 * @author xschen
 */
public class UserService {
    private static final String userServicePrefix = "/services/user";
    private static final String orderServicePrefix = "/services/order";

    public void userService(String ip) {
        CountDownLatch latch = new CountDownLatch(1);
        try {
            ZooKeeper zk = new ZooKeeper(Constants.connString, 5000, new Watcher() {
                @Override
                public void process(WatchedEvent event) {
                    latch.countDown();
                }
            });
            latch.await();
            System.out.println(System.currentTimeMillis() / 1000 + " UserService: " + ip + " 连接建立成功");

            // 把自己注册到服务注册中心
            startRegister(zk, ip);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 1. 尝试创建 /services/user/{ip} 节点
     *      如果创建成功，表示正常注册到服务中心
     *      如果创建失败，直接退出，去查IP被占用的原因
     *
     * 2. 如果正常注册到服务注册中心，由于用户服务需要调用订单服务，
     *   所以，需要监听订单服务，一旦订单服务的列表发生变化（新增，宕机），这里会及时感应到。
     * @param zk
     * @param ip
     */
    private void startRegister(ZooKeeper zk, String ip) {
        zk.create(userServicePrefix + "/" + ip, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL, new AsyncCallback.StringCallback() {
            @Override
            public void processResult(int rc, String path, Object ctx, String name) {
                if (rc == KeeperException.Code.OK.intValue()) {
                    System.out.println("userService " + ip + " 上线成功，Zookeeper的路径是：" + name);
                    try {
                        List<String> orderServiceIps = autoFindOrderService(zk, ip);
                        if (orderServiceIps.size() < 1) {
                            System.out.println("UserService 第一次" + ip + " 没有发现可用的 order 服务");
                        } else {
                            orderServiceIps.stream()
                                    .map(orderIp -> "UserService 第一次" + ip + "发现可用的 order 服务" + orderIp)
                                    .forEach(System.out::println);
                        }
                    } catch (Exception e) {
                        System.out.println("UserService 第一次" + ip + " 获取 order 服务异常");
                    }
                } else {
                    System.out.println("startRegister 状态异常：" + rc);
                }
            }

            private List<String> autoFindOrderService(ZooKeeper zk, String ip) throws KeeperException, InterruptedException {
                return zk.getChildren(orderServicePrefix, new Watcher() {
                    @Override
                    public void process(WatchedEvent event) {
                        /**
                         * 这个事件表示，子节点发生了变化，一旦发生了变化，
                         * 我们就需要获取最新的子节点列表，并且继续监听订单节点。
                         */
                        if (event.getType() == Event.EventType.NodeChildrenChanged) {
                            try {
                                List<String> orderServiceIps = autoFindOrderService(zk, ip);
                                if (orderServiceIps.size() < 1) {
                                    System.out.println("UserService " + ip + " 没有发现可用的Order服务");
                                } else {
                                    orderServiceIps.stream()
                                            .map(orderIp -> "UserService " + ip + "发现可用的 order 服务" + orderIp)
                                            .forEach(System.out::println);
                                }
                            } catch (Exception e) {
                                System.out.println("UserService " + ip + " 获取Order服务异常");
                            }
                        } else {
                            /**
                             * 我们目前只关注 子节点变化的事件，其他事件先不关注
                             */
                            System.out.println("其他事件："+event.getType().getIntValue()+event.getState().getIntValue());
                        }
                    }
                });
            }
        }, "callback data");
    }
}
